#include "ThreadPool.h"
#include "Bussiness.h"

// void cleanup(void* arg) {
//     ThreadPool* pool = (ThreadPool*)arg;
//     pthread_rwlock_unlock(&pool->rwlock);
// }

void* event_loop(void* arg) {
    ThreadPool* pool = (ThreadPool*) arg;
    pthread_t tid = pthread_self();
    // 线程清理函数
    //  pthread_cleanup_push(cleanup, pool);
    for(;;) {
        int connfd = blockq_pop(pool->task_queue); // 阻塞点
        if (connfd == -1) {
            printf("%lu exits\n", tid);
            pthread_exit(0);
        }
        
        printf("%lx: handle request from %d\n", tid, connfd);
        // 子进程处理业务逻辑
        send_file(connfd);
        sleep(10);

        printf("%lx: requst from %d done\n", tid, connfd);
    }          
    // pthread_cleanup_pop(0);
}

ThreadPool* create_thread_pool(int n)
{
    ThreadPool* pool = (ThreadPool*)malloc(sizeof(ThreadPool));
    pool->threads = (pthread_t*) calloc(n, sizeof(pthread_t));
    pool->num_threads = n;
    pool->task_queue = blockq_create();

    // 创建线程
    for(int i = 0; i < n; i++) {
        pthread_create(&pool->threads[i], NULL, event_loop, pool);
        printf("INFO: create a thread %lx\n", pool->threads[i]);
    }

    return pool;
}

void destroy_thread_pool(ThreadPool* pool) 
{
    free(pool->threads);
    blockq_destroy(pool->task_queue);
    free(pool);
}

int operation_puts(int netfd, char* pathname){
    train_t train;
    recv(netfd, &train.length, sizeof(train.length), MSG_WAITALL);
    recv(netfd, train.data, train.length, MSG_WAITALL);
    char filename[1024] = {0};
    memcpy(filename, train.data, train.length);

    char localpath[1024] = {0};
    sprintf(localpath, "%s%s", pathname, filename);

    int fd = open(localpath, O_RDWR | O_CREAT | O_EXCL, 0666);
    off_t filesize;

    recv(netfd, &train.length, sizeof(train.length), MSG_WAITALL);
    recv(netfd, train.data, train.length, MSG_WAITALL);
    memcpy(&filesize, train.data, train.length);
    printf("filesize = %ld\n", filesize);

    if(filesize >= BIG_FILE){
        ftruncate(fd, filesize);
        char* p = (char*)mmap(NULL, filesize, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
        recv(netfd, p, filesize, MSG_WAITALL);
        munmap(p, filesize);
    }else{
        while(1){
            recv(netfd, &train.length, sizeof(train.length), MSG_WAITALL);
            if(train.length == 0) break;
            recv(netfd, train.data, train.length, MSG_WAITALL);
            write(fd, train.data, train.length);
        }
    }
    printf("100.00%%\n");
    return 0;
}










